package com.atguigu.userprofile.app;

import com.atguigu.userprofile.common.bean.TagInfo;
import com.atguigu.userprofile.common.constants.CodeConst;
import com.atguigu.userprofile.common.dao.TagInfoDAO;
import com.atguigu.userprofile.common.util.MyClickhouseUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;

import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;

public class TaskBitmapApp {


    //1 读取mysql 中标签集合
    //
    //建表   每天建  是  一次性建立
    //      一次性手动建立 4张表 （区别，标签值字段类型不同）
    //      分区 每天每个标签一个分区    ，标签值作为排序字段
    //
    //2 读   select
    //
    //3 写   insert select
    public static void main(String[] args) {


        //任务4 实质上是有java程序组合sql ，交由clickhouse 计算完成，  这里使用spark上下文只是为了统一的提交方式 统一调度
        SparkConf sparkConf = new SparkConf().setAppName("task_bitmap_app");//.setMaster("local[*]");
        SparkContext sparkContext = new SparkContext(sparkConf);

        String busiDate = args[1];

        //1 读取mysql 中标签集合
        List<TagInfo> tagInfoList = TagInfoDAO.getTagInfoListWithOn();


        //2  要根据标签值类型把标签集合拆分成4份 ，分别插入不同的的表中
        List tagInfoListForString = new ArrayList<>();
        List tagInfoListForLong = new ArrayList<>();
        List tagInfoListForDecimal = new ArrayList<>();
        List tagInfoListForDate = new ArrayList<>();

        for (TagInfo tagInfo : tagInfoList) {
            if (tagInfo.getTagValueType().equals(CodeConst.TAG_VALUE_TYPE_LONG)) {
                tagInfoListForLong.add(tagInfo);
            } else if (tagInfo.getTagValueType().equals(CodeConst.TAG_VALUE_TYPE_DECIMAL)) {
                tagInfoListForDecimal.add(tagInfo);
            } else if (tagInfo.getTagValueType().equals(CodeConst.TAG_VALUE_TYPE_STRING)) {
                tagInfoListForString.add(tagInfo);
            } else if (tagInfo.getTagValueType().equals(CodeConst.TAG_VALUE_TYPE_DATE)) {
                tagInfoListForDate.add(tagInfo);
            }
        }
        // 根据标签集合插入到不同的bitmap表
        insertBitmap(tagInfoListForString,"user_tag_value_string",busiDate);
        insertBitmap(tagInfoListForLong,"user_tag_value_long",busiDate);
        insertBitmap(tagInfoListForDecimal,"user_tag_value_decimal",busiDate);
        insertBitmap(tagInfoListForDate,"user_tag_value_date",busiDate);



    }


    // 根据标签集合插入到不同的bitmap表
    private  static  void insertBitmap(List<TagInfo> tagInfoList,String targetTableName ,String busiDate ){
        if(tagInfoList.size()==0){
           return;
        }

        List<String> tagCodeList = tagInfoList.stream().map(tagInfo -> "('" + tagInfo.getTagCode().toLowerCase() + "'," + tagInfo.getTagCode().toLowerCase() + ") ").collect(Collectors.toList());
        String tagCodeSQL = StringUtils.join(tagCodeList, ",");
        String sourceTable ="user_tag_merge_"+busiDate.replace("-","");

        // 幂等性处理  清理数据
        //  alter table  $targetTableName   delete where dt='busidate'
        String deleteTable=" alter table  "+targetTableName+"  delete where dt='"+busiDate+"' ";
        System.out.println(deleteTable);
        MyClickhouseUtil.executeSql(deleteTable);


        // insert into $targetTableName
        //select   tv.1 tag_code,tv.2  tag_value,  groupBitmapState( cast(uid as UInt64) )  ,'$busiDate'
        //from (
        //select  uid, arrayJoin([('tg_person_base_gender',tg_person_base_gender) ,
        //    ('tg_person_base_agegroup',tg_person_base_agegroup)  $tagCodeSQL ])  tv
        //       from user_tag_merge_$busiDate ) tt
        //group by tv.1,tv.2

        String insertSQL="    insert into  " +targetTableName+
                "        select   tv.1 tag_code, if(tv.2='','0',tv.2)  tag_value,  groupBitmapState( cast(uid as UInt64) )  ,'"+busiDate+"'\n" +
                "         from (\n" +
                "         select  uid, arrayJoin([ "+tagCodeSQL+" ])  tv\n" +
                "              from "+sourceTable+" ) tt\n" +
                "         group by tv.1,tv.2";
        System.out.println(insertSQL);

        MyClickhouseUtil.executeSql(insertSQL);



    }
}
